package com.fulihui.wgj.biz.impl;


import com.fulihui.common.util.JedisUtil;
import com.fulihui.wgj.biz.SendMassMsgBiz;
import com.fulihui.wgj.biz.WxUserInfoBiz;
import com.fulihui.wgj.dal.impala.domain.WxUserInfoSend;
import com.fulihui.wgj.dal.mybatis.domain.WxMassMsgLog;
import com.fulihui.wgj.dal.mybatis.repository.WxSendTemplateLogRepository;
import com.fulihui.wgj.facade.msg.MassMsg;
import com.fulihui.wgj.facade.request.mass.CalcMassMsgRequest;
import com.fulihui.wgj.mq.kafka.producer.KafkaProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 图文群发业务
 *
 * @author wang_wx
 * @date 2018-08-01
 */
@Service
public class SendMassMsgBizImpl implements SendMassMsgBiz {

    private static final Logger LOGGER = LoggerFactory.getLogger(SendMassMsgBizImpl.class);

    @Autowired
    ThreadPoolTaskExecutor threadPool;

    @Autowired
    JedisUtil jedisUtil;

    @Autowired
    KafkaProducer kafkaProducer;

    @Autowired
    WxUserInfoBiz wxUserInfoBiz;

    @Autowired
    WxSendTemplateLogRepository wxSendTemplateLogRepository;

    @Override
    public void sendMassMsg(CalcMassMsgRequest request, WxMassMsgLog task) {
        LOGGER.info("sendMassMsg start .......");
        Integer recordCount = Math.toIntExact(task.getTotalCount());
        if (recordCount == null || recordCount == 0) {
            recordCount = Math.toIntExact(wxUserInfoBiz.queryCount(request));
        }
        //异步执行
        Integer finalRecordCount = recordCount;
        LOGGER.info("#####发送图文群发消息的发送总数是" + finalRecordCount);
        threadPool.execute(() -> {
                    try {
                        MassMsg msg = new MassMsg();
                        msg.setRecordId(task.getId());
                        msg.setAppId(task.getAppId());
                        msg.setMediaId(task.getMediaId());
                        msg.setInfoType(task.getMsgType());

                        //分页查询
                        int pageSize = 5000;
                        long pageCount = (finalRecordCount + pageSize - 1) / pageSize;
                        LOGGER.info(String.format("#####发送图文群发消息的发送总数是：%d 发送的分页是：%d 发送的批次是 %d发送的公众号是：%s", finalRecordCount, pageCount, task.getId(), request.getAppId()));
                        String nextOpenid = null;
                        //发送总条数
                        for (int pageNum = 1; pageNum <= pageCount; pageNum++) {
                            if (pageNum == pageCount) {
                                msg.setLast(Boolean.TRUE);
                            }
                            //分批取出用户信息
                            request.setPage(pageNum);
                            request.setRows(pageSize);
                            List<WxUserInfoSend> userInfoSends = wxUserInfoBiz.queryPage(request, nextOpenid);

                            //缓解查询压力，每批休眠
                            TimeUnit.SECONDS.sleep(5);

                            if (!CollectionUtils.isEmpty(userInfoSends)) {
                                List<String> openIds = userInfoSends.stream().map(WxUserInfoSend::getOpenid).collect(Collectors.toList());
                                msg.setOpenIds(openIds);
                                kafkaProducer.sendMassMsg(msg);
                            }
                            nextOpenid = userInfoSends.get(userInfoSends.size() - 1).getOpenid();
                        }
                    } catch (Exception e) {
                        LOGGER.error("发送图文群发消息错误,:e{}", e);
                    }
                }
        );
    }
}
